package com.gugee.tiktok.data.consumer.keyword.core.main;

import cn.hutool.core.date.DateUtil;
import com.google.common.util.concurrent.AtomicDouble;
import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.tiktok.common.model.dal.blogger.AwemeCommentRepository;
import com.gugee.tiktok.common.model.dal.blogger.AwemeRepository;
import com.gugee.tiktok.common.model.es.ESAweme;
import com.gugee.tiktok.common.model.es.Keyword;
import com.gugee.tiktok.common.model.mysql.AwemeComment;
import com.gugee.tiktok.data.common.msg.KeywordMessage;
import com.gugee.tiktok.data.consumer.keyword.config.KeywordAppConfig;
import com.gugee.tiktok.data.consumer.keyword.core.spider.KeywordSpiderClient;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;

@Slf4j
@Component
public class SaveAwemeKeywordMain   {

    private final String AWEME_INDEX = "aweme_index_";
    @Autowired
    AwemeRepository awemeRepository;
    @Autowired
    RestHighLevelClient highLevelClient;
    @Autowired
    KeywordSpiderClient keywordSpiderClient;
    @Autowired
    AwemeCommentRepository awemeCommentRepository;
    @Autowired
    ESAweme.ESAwemeMapper esAwemeMapper;
    ForkJoinPool forkJoinPool;
    @Autowired
    ScheduleTemplate scheduleTemplate;
    @Autowired
    KeywordAppConfig keywordAppConfig;


    private String workName = "keyword-worker";
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(keywordAppConfig.getKeywordPollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    public void handle(List<KeywordMessage> keywordMessages) throws ExecutionException, InterruptedException {

            forkJoinPool.submit(() -> keywordMessages.parallelStream().forEach(msg -> {
                try {
                    run(msg);
                } catch (Exception ex) {
                    log.error("keyword error. awemeId:{},{}" , msg.getOAwemeId(),ex.getMessage());
                }
            })).get();
    }
    public void run(KeywordMessage keywordMessage) {
        long awemeId=keywordMessage.getOAwemeId();
        List<AwemeComment> awemeComments=keywordMessage.getAwemeComments();
        if(keywordMessage.getId()==null || keywordMessage.getId()==0){
            log.error("接收到的消息无id值，OAwemeId={},CommentsSize={}",awemeId,awemeComments.size());
            return;
        }


        if(awemeComments==null ||awemeComments.size()==0){
            log.error("可能oss无评论，OAwemeId={},CommentsSize={}",awemeId,awemeComments.size());
            return;
        }


        //todo 提需求，让对方提供多视频粒度的接口，这样可批量解析，大大减少请求量
        List<Keyword> keywordList = getKeywordByAi(awemeComments,awemeId);
        if(keywordList.size() == 0){
            log.error("ai分析关键词为0，OAwemeId={},CommentsSize={},getKeywordsByAi=0",awemeId,awemeComments.size());
            return;
        }

            UpdateRequest updateRequest = new UpdateRequest();
            updateRequest.index(AWEME_INDEX+getDateStr(keywordMessage.getOCreateTime()));
            updateRequest.id(String.valueOf(keywordMessage.getId()));
            Map<String,List<Keyword>> map = new HashMap<>();
            map.put("awemeTags",keywordList);
            updateRequest.doc(JsonUtil.parseJson(map),XContentType.JSON);
            try {
                //todo 需批量保存提高效率
                highLevelClient.update(updateRequest,RequestOptions.DEFAULT);
                awemeRepository.updateAwemeAnalysisCommentsByoAwemeId(System.currentTimeMillis()/1000,"keywordAnalysis",awemeComments.size(),keywordMessage.getOAwemeId());

            }catch (Exception e) {
                log.error("OAwemeId={},CommentsSize={},Exception={}",awemeId,awemeComments.size(),e.getMessage());
            }finally {

            }
    }


    private String getDateStr(long second){
        return DateUtil.format(DateUtil.calendar(second*1000).getTime(),"yyyyMM");
    }

    public List<String> keywordAi(List<AwemeComment> awemeComments){
        if(awemeComments==null ||awemeComments.size()==0){
            return new ArrayList<>();
        }
        /**
         * 一个批次的大小
         */
        int size=keywordAppConfig.getAnalysisKeywordComments();
        /**
         * 计算批次
         */
        int batchs=(awemeComments.size()+size-1)/size;

        int lastBatchs=awemeComments.size()%size;

        Set<String> rst=new HashSet<>();
        for(int i=0;i<batchs;i++){
            List<AwemeComment> comments;
            if(lastBatchs>0 && (batchs-i)==1){
                comments=awemeComments.subList(i*size,awemeComments.size());
            }else{
                comments=awemeComments.subList(i * size, (i + 1) * size);
            }
//            log.info("keywordSpiderClient comments size/awemeComments:{}/{}", comments.size(),awemeComments.size());
            Map<Integer,String> requestBody = buildRequestBody(comments);

             long start = System.currentTimeMillis();
             List<String> keywordList =   keywordSpiderClient.getCommentAnalyzed(requestBody);
             if((System.currentTimeMillis()-start)/1000>keywordAppConfig.getDebugTimeLimit()) {
                    log.info("keywordSpiderClient  use time:" + (System.currentTimeMillis() - start) / 1000 + "s");
             }
             rst.addAll(keywordList);
        }
        return new ArrayList<>(rst);

    }
    /**
     * 从AI接口获取分词
     * @param awemeComments
     * @return
     */
    private List<Keyword> getKeywordByAi(List<AwemeComment> awemeComments ,long awemeId){
        List<String> commentAnalyzed = null;
        List<Keyword> keywords = new ArrayList<>();

        try{
            long start = System.currentTimeMillis();
            commentAnalyzed = keywordAi(awemeComments);
            if((System.currentTimeMillis()-start)/1000>keywordAppConfig.getDebugTimeLimit()) {
                log.info("SaveAwemeKeywordEvent CommentAnalyzed use time:{}s,commentsize={}" , (System.currentTimeMillis() - start) / 1000 ,awemeComments.size());
            }
        } catch (Exception e){
//            e.printStackTrace();
            log.error("AwemeId={},commentsize={},SaveAwemeKeywordEvent CommentAnalyzed error! {} ",awemeId,awemeComments.size(), e.getMessage());
        }

        if(commentAnalyzed==null || commentAnalyzed.size()==0 ){
            return keywords;
        }
        Map<String,Integer> rels = new HashMap<>();

        AtomicDouble keywordCount = new AtomicDouble();

        for (String l : commentAnalyzed) {
            if(null == rels.get(l)){
                int count = Collections.frequency(commentAnalyzed,l);
                rels.put(l,count);
                keywordCount.addAndGet(count);
            }
        };

        rels.forEach((k,v)->{
            Keyword keyword = new Keyword();
            keyword.setKeyword(k);
            keyword.setPercent(((int)(v/keywordCount.get()*100*100))/100.0);
            keywords.add(keyword);
        });


        if(keywords.size() > 1){
            Collections.sort(keywords,(first,second)-> {
                if(first.getPercent() > second.getPercent()){
                    return -1;
                } else if (first.getPercent() == second.getPercent()){
                    return 0;
                }
                return 1;
            });
        }

        return keywords.size() >10?keywords.subList(0,10):keywords;
    }

//    private List<Keyword> getKeyword(long awemeId){
//        Map<Integer,String> requestBody = getRequestBody(awemeId);
//        List<Keyword> keywords = new ArrayList<>();
//
//        if(requestBody.isEmpty()){
//            return keywords;
//        }
//        return getKeywordByAi(requestBody,null);
//    }

//    private List<ESAweme> getAwemeListByEs(long uid){
//        SearchRequest searchRequest = new SearchRequest(getEsIndex());
//        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
//        searchSourceBuilder.query(QueryBuilders.matchQuery("uid",uid));
//        searchSourceBuilder.size(50);
//        searchRequest.source(searchSourceBuilder);
//        List<ESAweme> esAwemes = new ArrayList<>();
//        try {
//            SearchResponse searchResponse = highLevelClient.search(searchRequest,RequestOptions.DEFAULT);
//            val hits = searchResponse.getHits().getHits();
//
//            for (SearchHit hit : hits) {
//                val aweme = JsonUtil.jsonParse(ESAweme.class, hit.getSourceAsString());
//                esAwemes.add(aweme);
//            }
//
//        } catch (IOException e) {
//            log.error("SaveAwemeKeywordEvent error! bloggerId:",uid, e);
//        }
//        return esAwemes;
//    }

//    private List<ESAweme> getAwemeListByEsAndTagsIsNull() {
//        SearchRequest searchRequest = new SearchRequest(getEsIndex());
//        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
//        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
//        //boolQueryBuilder.minimumShouldMatch(3);
//        boolQueryBuilder.must(QueryBuilders.rangeQuery("commentCount").gt(50));
//
//
//
//        BoolQueryBuilder boolQueryBuilder2 = new BoolQueryBuilder();
//        boolQueryBuilder2.must(QueryBuilders.existsQuery("awemeTags"));
//
//        boolQueryBuilder.mustNot(QueryBuilders.nestedQuery("awemeTags",boolQueryBuilder2, ScoreMode.Total));
//        searchSourceBuilder.sort("createTime", SortOrder.DESC);
//        searchSourceBuilder.from(0);
//        searchSourceBuilder.size(500);
//        searchSourceBuilder.query(boolQueryBuilder);
//        searchRequest.source(searchSourceBuilder);
//        List<ESAweme> esAwemes = new ArrayList<>();
//        try {
//            SearchResponse searchResponse = highLevelClient.search(searchRequest,RequestOptions.DEFAULT);
//            val hits = searchResponse.getHits().getHits();
//
//            for (SearchHit hit : hits) {
//                val aweme = JsonUtil.jsonParse(ESAweme.class, hit.getSourceAsString());
//                esAwemes.add(aweme);
//            }
//
//        } catch (IOException e) {
//            log.error("SaveAwemeKeywordEvent TagsIsNull error! ", e.getMessage());
//        }
//        return esAwemes;
//
//    }

    private String getEsIndex(){
        LocalDateTime now = LocalDateTime.now();
        StringBuffer stringBuffer = new StringBuffer(AWEME_INDEX+DateUtil.format(now,"yyyyMM")).append(",");
        stringBuffer.append(AWEME_INDEX+DateUtil.format(now.minusMonths(1),"yyyyMM")).append(",");
        stringBuffer.append(AWEME_INDEX+DateUtil.format(now.minusMonths(2),"yyyyMM"));
        return stringBuffer.toString();
    }


    private String getCurMonthIndex(){
        LocalDateTime now = LocalDateTime.now();
        StringBuffer stringBuffer = new StringBuffer(AWEME_INDEX+DateUtil.format(now,"yyyyMM"));
        return stringBuffer.toString();
    }


    private Map<Integer,String> getRequestBody(long awemeId){
        long start = System.currentTimeMillis();
        Map<Integer,String> body = new HashMap<>();

        try{
            String comment = "";
            val aweme = awemeRepository.findByoAwemeId(awemeId);
            if(aweme.isPresent()){
                comment = awemeCommentRepository.get(awemeId,aweme.get().getDatecode());
            }
            if((System.currentTimeMillis()-start)/1000>5)
                log.info("ESRefreshAwemeEventListener get oss data awemeId:{} use time:{}s",awemeId,(System.currentTimeMillis()-start)/1000);
            List<AwemeComment> comments = new ArrayList<>();
            comments = JsonUtil.jsonParse(comments.getClass(),comment);

            return buildRequestBody(comments);
        }catch (Exception e){
            log.error("ESRefreshAwemeEventListener get oss data error!",e.getMessage());
        }

        return body;

    }

    private Map<Integer,String> buildRequestBody(List<AwemeComment> comments){
        Map<Integer,String> body = new HashMap<>();

        for (int i = 0; i < comments.size(); i++) {
            body.put(i,comments.get(i).getText());
        }

        return body;
    }


}
